﻿/*
   Copyright 2013 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Serialization;
using System.Security.Principal;
using System.Text;
using System.Threading;
using Flower.Client;
using Flower.Logging;
using Flower.Services.Data;
using Flower.Testing;
using Flower.Workflow;
using NUnit.Framework;

namespace Flower.Processor.Tests
{
    [TestFixture]
    public class ProcessorTest : IProcessingManager
    {
        private static readonly Log Log = LogManager.CreateLog(typeof(ProcessorTest).Name);

        private const int PROCESSES_COUNT = 10;

        public class BlockingService
        {
            public void BlockingMethod()
            {
                Thread.Sleep(100);
            }
        }

        [DataContract]
        public class TestWorkflow : Workflow.Process, IWorkflow<object, object>
        {
            public void DefineLocalSets(ISetsBuilder bld)
            {
            }

            public object Initialize(IInitializationContext context, object arg)
            {
                return null;
            }

            public void DefineProcess(IProcessBuilder bld)
            {
                bld
                    .Exec(ctx => { /* Do nothing. */ })
                    .Invoke<BlockingService>
                    (
                        ctx => "/Services/TestServices/MyService", 
                        (ctx, svc) => svc.BlockingMethod()
                    )
                    .Exec(ctx => { /* Do nothing. */ });
            }
        }

        [DataContract]
        [AllowExecuteTo("Processor/Specific")]
        public class TestWorkflowRestrictedSpecific : Workflow.Process, IWorkflow<object, object>
        {
            public void DefineLocalSets(ISetsBuilder bld)
            {
            }

            public object Initialize(IInitializationContext context, object arg)
            {
                return null;
            }

            public void DefineProcess(IProcessBuilder bld)
            {
                bld
                    .Exec(ctx => { /* Do nothing. */ })
                    .Invoke<BlockingService>
                    (
                        ctx => "/Services/TestServices/MyService",
                        (ctx, svc) => svc.BlockingMethod()
                    )
                    .Exec(ctx => { /* Do nothing. */ });
            }
        }

        [DataContract]
        [AllowExecuteTo("Processor/Another")]
        public class TestWorkflowRestrictedAnother : Workflow.Process, IWorkflow<object, object>
        {
            public void DefineLocalSets(ISetsBuilder bld)
            {
            }

            public object Initialize(IInitializationContext context, object arg)
            {
                return null;
            }

            public void DefineProcess(IProcessBuilder bld)
            {
                bld
                    .Exec(ctx => { /* Do nothing. */ })
                    .Invoke<BlockingService>
                    (
                        ctx => "/Services/TestServices/MyService",
                        (ctx, svc) => svc.BlockingMethod()
                    )
                    .Exec(ctx => { /* Do nothing. */ });
            }
        }

        [DataContract]
        public class TestWorkflowWaiting : Workflow.Process, IWorkflow<object, object>
        {
            public void DefineLocalSets(ISetsBuilder bld)
            {
            }

            public object Initialize(IInitializationContext context, object arg)
            {
                return null;
            }

            public void DefineProcess(IProcessBuilder bld)
            {
                bld
                    .Exec(ctx => { /* Do nothing. */ })
                    .Invoke<BlockingService>
                    (
                        ctx => "/Services/TestServices/MyService",
                        (ctx, svc) => svc.BlockingMethod()
                    )
                    .Enqueue(ctx => "/Sets/Shared/Waiting", ctx => 1, _ => MessageFormat.BinXml, ctx => {})
                    .Dequeue<int>(ctx => "/Sets/Shared/Continue", (ctx, msg) => {})
                    .Exec(ctx => { /* Do nothing. */ });
            }
        }

        private readonly ManualResetEventSlim _processFinished = new ManualResetEventSlim();
        private readonly List<string> _finished = new List<string>();
        private readonly List<string> _broken = new List<string>();
        private int _processesStarted;

        [TestFixtureSetUp]
        public void Setup()
        {
            string identity = WindowsIdentity.GetCurrent().Name;

            new DirectoryBuilder(ApplicationContext.GetDirectoryClient().Directory)
                .Root("/Services")
                    .ServiceContainer
                    (
                        "TestServices",
                        "<objects>" +
                        "  <object id=\"MyService\" type=\"" + typeof(BlockingService).AssemblyQualifiedName + "\" />" +
                        "</objects>"
                    ).End()
                .End()
                .Root("/Workflows")
                    .Workflow("TestWorkflow", typeof(TestWorkflow).AssemblyQualifiedName, null).End()
                    .Workflow("TestWorkflowRestrictedSpecific", typeof(TestWorkflowRestrictedSpecific).AssemblyQualifiedName, null).End()
                    .Workflow("TestWorkflowRestrictedAnother", typeof(TestWorkflowRestrictedAnother).AssemblyQualifiedName, null).End()
                    .Workflow("TestWorkflowWaiting", typeof(TestWorkflowWaiting).AssemblyQualifiedName, null).End()
                .End()
                .RegisterProcessor("TestProcessor")
                .Root("/Sets/Shared")
                    .Set("Waiting", typeof(int).AssemblyQualifiedName, 0).End()
                    .Set("Continue", typeof(int).AssemblyQualifiedName, 0).End()
                .End()
                .Root("/Roles/Processor")
                    .Role("Specific", new[] { identity }).End()
                .End();
        }

        [Test]
        public void ExecuteRestrictedProcess()
        {
            var host = new ProcessorHost();
            var client = ApplicationContext.GetDirectoryClient();

            _broken.Clear();
            _finished.Clear();
            _processFinished.Reset();
            _processesStarted = 0;

            host.Start(this);

            try
            {
                var pids = new List<string> 
                {
                    client.StartProcess<object, object>("/Workflows/TestWorkflowRestrictedSpecific", null, Log).Pid,
                    client.StartProcess<object, object>("/Workflows/TestWorkflowRestrictedAnother", null, Log).Pid
                };

                _processesStarted = pids.Count;

                foreach (var pid in pids)
                {
                    client.Directory.Move(pid, "/Processors/TestProcessor/Running");
                    host.LoadAndExecuteProcess(pid);
                }

                if (!_processFinished.Wait(TimeSpan.FromSeconds(5)))
                {
                    Assert.Fail("The process is running for too long.");
                }

                Assert.IsTrue(_broken.Count == 1);
                Assert.IsTrue(_finished.Count == 1);
            }
            finally
            {
                host.Stop();
            }
        }

        [Test]
        public void LoadAndExecuteProcess()
        {
            var host = new ProcessorHost();
            var client = ApplicationContext.GetDirectoryClient();

            _broken.Clear();
            _finished.Clear();
            _processFinished.Reset();
            _processesStarted = 0;

            host.Start(this);

            try
            {
                var pids = new List<string>();

                for (int i = 0; i < PROCESSES_COUNT; i++)
                {
                    pids.Add(client.StartProcess<object, object>("/Workflows/TestWorkflow", null, Log).Pid);
                    _processesStarted++;
                }

                var rng = new Random();

                foreach (var pid in pids)
                {
                    client.Directory.Move(pid, "/Processors/TestProcessor/Running");
                    host.LoadAndExecuteProcess(pid);

                    Thread.Sleep(TimeSpan.FromMilliseconds(rng.Next(30)));
                }
                
                if (!_processFinished.Wait(TimeSpan.FromSeconds(10)))
                {
                    Assert.Fail("The process is running for too long.");
                }

                Assert.IsEmpty(_broken);
            }
            finally
            {
                host.Stop();
            }
        }

        [Test]
        public void RunService()
        {
            var service = new ProcessorService();
            var client = ApplicationContext.GetDirectoryClient();

            service.DoStart(ProcessorService.Options.DoNotCastAppDomains | ProcessorService.Options.DoNotStartServiceHost);

            try
            {
                var pids = new List<string>();

                for (int i = 0; i < PROCESSES_COUNT; i++)
                {
                    pids.Add(client.StartProcess<object, object>("/Workflows/TestWorkflowWaiting", null, Log).Pid);
                }
                
                service.CheckPendingProcesses();

                var limit = 10;
                
                do
                {
                    Thread.Sleep(5000);
                    --limit;
                } while
                (
                    limit > 0 &&
                    client.Directory.CountChildren("/Sets/Shared/Waiting", DirectoryEntryTypes.Message) < PROCESSES_COUNT
                );

                if (limit == 0)
                {
                    Assert.Fail();
                }

                for (int i = 0; i < PROCESSES_COUNT; i++)
                {
                    client.PutMessage("/Sets/Shared/Continue", null, 1, BlobFormat.BinXml);
                }

                limit = 10;

                do
                {
                    Thread.Sleep(5000);
                    --limit;
                } while 
                (
                    limit > 0 &&
                    client.Directory.CountChildren("/Processors/TestProcessor/Finished", DirectoryEntryTypes.Process) < PROCESSES_COUNT
                );

                if (limit == 0)
                {
                    Assert.Fail();
                }
            }
            finally
            {
                service.DoStop();
            }
        }

        #region IProcessingManager members

        public void MarkFinished(string pid)
        {
            ApplicationContext
                .GetDirectoryClient()
                .Directory
                .Move(pid, "/Processors/TestProcessor/Finished");

            lock (_finished)
            {
                _finished.Add(pid);

                if (_broken.Count + _finished.Count == _processesStarted)
                {
                    _processFinished.Set();
                }
            }
        }

        public void MarkBroken(string pid)
        {
            ApplicationContext
                .GetDirectoryClient()
                .Directory
                .Move(pid, "/Processors/TestProcessor/Broken");

            lock (_broken)
            {
                _broken.Add(pid);

                if (_broken.Count + _finished.Count == _processesStarted)
                {
                    _processFinished.Set();
                }
            }
        }

        public void MarkWaiting(string pid)
        {
        }

        public void MarkRunning(string pid)
        {
        }

        public void MarkSuspended(string pid, DateTime until, string linkId)
        {
        }

        #endregion
    }
}
